package com.atguigu.gmall.item.service.impl;

import com.atguigu.gmall.common.cache.BloomName;
import com.atguigu.gmall.common.cache.GmallCache;
import com.atguigu.gmall.common.constant.RedisConst;
import com.atguigu.gmall.feign.product.SkuInfoFeignClient;
import com.atguigu.gmall.feign.search.SkuEsFeignClient;
import com.atguigu.gmall.item.service.ItemService;
import com.atguigu.gmall.model.product.BaseCategoryView;
import com.atguigu.gmall.model.product.SkuInfo;
import com.atguigu.gmall.model.product.SpuSaleAttr;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 要使用缓存：
 * 1、null结果也应该缓存：缓存穿透
 * 2、缓存的数据都加上过期时间，防止缓存没有被代码更新，数据一直是错误的
 * 3、查询数据库要加锁
 * 4、把所有数据的失效时间设置成随机的【不用做】  30min； 只要加了失效时间就行不用随机
 * <p>
 * 着重注意两点：
 * 1、加锁（解决击穿、雪崩（录入数据的时间本就不一样，时间维度扩散的，只需要解决击穿）？）
 * 2、数据有过期时间（数据过期一定要加）
 * <p>
 * 哪些数据放到缓存？
 * 1、查多改少【菜单、sku、spu、优惠券】
 * 2、热点（查多）
 * 3、一致性要求不高【指不用强一致，缓存和数据库的数据是弱一致的】
 * 强一致： 分布式系统 Raft, 数据改完以后，都是同步的。MySQL - Redis 同步的
 * 弱一致： 最终一致性。redis的数据最终和mysql的是一致的就行
 * 不适合放缓存？
 * 1、改多
 * 和钱相关
 * 库存（实时校验库存）
 * sku的销量saleCount？ 销量即使数据库变了，缓存没有更新过来无所谓
 * sku数据在缓存中有。看销量（对用户不重要，每个月进行账单统计的时候才需要精确查）？
 * 3、数据一致性问题：
 * 数据库修改以后，缓存的数据需要同步过来；【缓存中的所有数据都应当有过期时间】
 * 1）、缓存的每个数据都必须有过期时间？（允许有一段时间数据不一致）【最终一致性】
 * sku:info:50：后台改了50号数据。  30min 分钟过期，
 * 数据过期以后，下一次对此数据的查询会触发更新逻辑。最终缓存中还是新数据。
 * <p>
 *      双写模式：写数据库+写缓存；
 *              本来要同步缓存（改数据的同时给缓存存一份）；
 *      失效模式：写数据库+删缓存；
 *              下一次查询，由于缓存中没有，自己会查数据库，然后又更新到缓存
 * <p>
 * 这个两种。都可以：99% 代码都是运行成功的。只要成功就能改掉数据。
 * 如果失败，就可以寄希望于 30min 分钟过期。
 * <p>
 * 2）、一劳永逸的解决方案
 * 核心：【把数据库的最后一次修改放到缓存】
 * 2.1）、Canal：
 * 安装Canal；开发同步到binlog以后的代码。 sku_info。
 * sku_info_update_5；   拿到5号记录放到缓存
 * 2.2）、mq：
 * 业务操作改了数据库，给mq发消息。
 * id:5   value: xxxx   updateTime: 获取到数据库的修改时间   发给mq
 * 有个线程专门订阅mq消息。  把数据放到缓存中。mq消息乱序
 * 缓存中此记录的时间如果大于我接下要给缓存中放的数据时间。我就放弃此次操作
 * <p>
 * 总结：
 * 面试官：你们怎么用缓存的，什么是缓存一致性问题，你们怎么解决
 * 答：
 * 1）、我们缓存的数据是实时性一致性要求不高（可以容忍一段时间不一致）
 * 2）、缓存一致性问题： 缓存中的数据没有同步到数据库最新的数据
 * 3）、我们业务代码是 失效模式+缓存数据过期时间。
 * 4）、如果失效模式导致了失败与错误数据，会在一段时间后 数据过期，下一次查询触发查询最新的
 * 面试官：怎么保证大并发下，缓存没有，查到的数据也是能放到缓存？怎么解决缓存击穿问题、缓存雪崩问题
 * 1）、加锁；有没有必要用分布式锁？？？？
 *      分布式锁性能太低了。。。
 *      直接操作：4000
 *      缓存没有要争分布式锁：50/s
 *      缓存中有：600/s
 *      最终决定：只加本地锁。
 * <p>
 * <p>
 * 4、锁的粒度问题
 * 1）、锁的粒度越细越快、
 * <p>
 * 5、分布式数据一致性的问题
 * raft： （强一致）【所有的操作都会通过领导】
 * 1,2,3
 * A----1-----把x号记录改成了y----返回成功或者失败
 *                          返回成功： 数据库（不是MySQL）集群大多数节点都已经同步了就会给你说成功
 * <p>
 * B----2-----获取x号记录-----领导会告诉还没有记录
 *                                  B----2---获取x记录，领导会返回记录
 * <p>
 * 问题：
 * mysql：后台同步
 * 写： 连上master；
 * master上改的东西一定不是立即读出来
 * 读： 连上slaver
 * 大概在1s左右读来
 * <p>
 * mysql：（【不要纠结，数据要立刻同步？】量子纠缠？）
 *      1、cpu在1ms的时候给mysql提交保存数据
 *      2、   mysql需要3ms才保存完
 *      3、cpu在2ms的时候要mysql此条数据
 *      4、   mysql肯定返回没有
 * <p>
 * 缓存：   失效模式+数据过期+本地锁+？？？
 * 缓存穿透攻击：
 *      查一个不存在的值，缓存没有总查数据库。允许缓存null值。
 *      随机值攻击：（缓存穿透攻击）
 *          1）、/api/item/50   1~10000
 *          2）、/api/item/时间戳变成数字
 * 核心思想： 缓存中有的，基本上就是数据库有的。
 *      布隆过滤器：
 *          1）、准备一个布隆过滤器
 *          2）、每次我们给数据库里面放了什么，告诉布隆
 *          3）、以后查数据
 *              1）、先问布隆过滤器，数据有没有。
 *                  1.1）、布隆说有，不一定有。
 *                          查缓存，查数据库，缓存null【利用布隆说有的进行null攻击】，请求达不到数据库
 *                  1.2）、布隆说没有就一定没有。直接返回null，缓存都不用看
 *
 */
@Slf4j
@Service
public class ItemServiceImpl implements ItemService {
    @Autowired
    RedissonClient redissonClient;

    //临时保存数据，本地缓存
    // redis.set(key,value)  Hash O(1)
    // mysql       树   O(logN)
//    private Map<String,Object> cache = new HashMap<>();

    @Autowired
    SkuInfoFeignClient skuInfoFeignClient;

    @Autowired
    StringRedisTemplate stringRedisTemplate;

    @Qualifier(BloomName.SKU)
    @Autowired
    RBloomFilter<Object> skuFilter;

    @Qualifier("corePool")
    @Autowired
    ThreadPoolExecutor executor;

    @Autowired
    SkuEsFeignClient skuEsFeignClient;

    /**
     * 1、分布式锁版的获取商品详情
     *      1）、锁有自动过期时间，防止死锁
     *      2）、加锁解锁都是原子的
     *      3）、锁的续期  10s，要给他不断续期，业务不中断。锁要自动延长时间？
     *          后台启动一个daemon线程，每隔5秒，让这个锁重新开始倒计时，续期就是直接续满。
     *          new Thread()
     *          new TimerTask(); 10
     *
     *      redisson： 基于redis做的分布式锁，分布式对象
     *
     * 如何做一个
     *   - 可重入、自旋、自动续期、分布式锁
     * redisson：  redis+son
     * redis：中间件
     * redisson：操作redis的客户端， 比：jedis、stringRedisTemplate强大
     * @param skuId
     * @return
     */

    @SneakyThrows  //自己设置的使用原生redis操作实现的分布式可重入锁业务
    public Map<String, Object> getSkuInfoWithRedisLock01(Long skuId,Boolean locked,String beforetToken) {
        // while ()  不能上来就用锁
        //1、先判断缓存中是否存在 单位时间内一个人的所有都执行完了
        Map<String, Object> cache = queryFromCache(skuId);
        if(cache != null){
            //2、缓存中有用缓存的
            log.info("缓存命中.....");
            return cache;
        }else {
            //3、缓存中没有调用业务逻辑真正查询
            //3.1）、为了不全放给数据库，占锁来查数据库
            log.info("缓存不命中，开始抢锁.....");
            //第一次连接告诉redis，占坑  用 "lock"
            //底层调用 set lock 1 NX EX 20
            String token =  beforetToken;
            Boolean aBoolean = locked ;
            if(!locked){  //判断之前有人已经帮我锁定了，我就直接用别人的锁
                token = UUID.randomUUID().toString();  //自己加锁就是新token
                //重入锁的设计
                aBoolean = stringRedisTemplate.opsForValue().setIfAbsent("lock", token,20,TimeUnit.SECONDS);
            }
            //还没运行断电了, 不可重入锁会死锁。  未来设计的所有锁都应该是可重入
            if (aBoolean){
                //第二次 才设置过期时间
                stringRedisTemplate.expire("lock",20, TimeUnit.SECONDS);
                //缓存中没有，并且占锁成功才执行业务
                //3.2）、占锁成功
                try {
                    log.info("抢占成功.....");
                    cache = getFromServiceItemFeign(skuId);
//                Thread   sleep是线程阻塞操作
                    //4、查询到数据后放入缓存
                    saveToCache(cache, skuId);  //异常
                } finally {
                    //5、删除锁，一定得执行
                    //保证业务正常出现了异常，finally兜底
                    //断电这种故障，redis过期时间,让redis自己删除\
                    //下面的删锁代码必须是原子型，否则可能出问题
//                    String lock = stringRedisTemplate.opsForValue().get("lock");
//                    if(token.equals(lock)){
//                        //删除我自己的锁
//                        stringRedisTemplate.delete("lock");
//                        System.out.println("删除分布式锁....");
//                        log.info("删除分布式锁....");
//                    }

                    //告诉redis，看 KEYS 是否等于 ARGV（argValue） ，如果是就删除 KEYS，否则返回0
                    String script = "if redis.call('get', KEYS[1]) == ARGV[1] " +
                            "then return redis.call('del', KEYS[1]) else return 0 end";
                    //lua脚本原子执行
                    stringRedisTemplate.execute(new DefaultRedisScript<>(script), Arrays.asList("lock"),token);

                }
//                try{
//                    //业务代码  1000
//                }finally {
//                    stringRedisTemplate.delete("lock");
//                }

            }else {
                //3.3）、占锁失败
                //??? 不断的操作redis    非公平锁，没有占锁成功就一直抢占
                // stringRedisTemplate.opsForValue().setIfAbsent("lock", "aaa")
                // stringRedisTemplate.opsForValue().setIfAbsent("lock", "aaa") == true  //生产的写法
                //stringRedisTemplate.opsForValue().setIfAbsent("lock", "aaa") == false
                // 一直抢直到占到锁，写死while(true)
                log.info("没抢成功，开始尝试自旋占锁.....");
                while (true) {
                    log.info("自旋中.....");
                    Thread.sleep(200);  //同一线程里面锁应该直接使用
                    String token2 = UUID.randomUUID().toString();
                    Boolean absent = stringRedisTemplate.opsForValue().setIfAbsent("lock", token2);
                    if(absent){
                        log.info("自旋占锁成功....."); //while(true) 自己调用自己，立马栈溢出
                        //自旋锁成功才调用自己, 栈溢出的写法，但是能看出效果
                        //这个方法能进去抢占成功，是因为absent已经超时了，redis删了
                        //锁设计为可重入锁
                        cache = getSkuInfoWithRedisLock01(skuId,absent,token2);
                        //释放锁  cpu速度远大于 redis，请用原子删锁和原子加锁

                        String script = "if redis.call('get', KEYS[1]) == ARGV[1] " +
                                "then return redis.call('del', KEYS[1]) else return 0 end";
                        //lua脚本原子执行
                        stringRedisTemplate.execute(new DefaultRedisScript<>(script), Arrays.asList("lock"),token2);
                        return cache;
                    }
                }

            }
        }
        return cache;
    }

    @SneakyThrows
    void saveToCache(Map<String, Object> data, Long skuId) {
        ObjectMapper objectMapper = new ObjectMapper();
        String string = objectMapper.writeValueAsString(data);
        ValueOperations<String, String> operations = stringRedisTemplate.opsForValue();
        //null值也要缓存
        operations.set("sku:info:" + skuId, string);
    }

    /**
     * 查缓存
     *
     * @param skuId
     * @return
     */
    @SneakyThrows
    public Map<String, Object> queryFromCache(Long skuId) {
        ObjectMapper mapper = new ObjectMapper();
        ValueOperations<String, String> operations =
                stringRedisTemplate.opsForValue();
        String redisContent = operations.get("sku:info:" + skuId);
        if (StringUtils.isEmpty(redisContent)) {
            return null;
        } else {
            return mapper.readValue(redisContent, new TypeReference<Map<String, Object>>() {
            });
        }
    }


    /**
     * 1、redisson会死锁吗？
     *      不会
     *      1、redisson 是可重入锁（同一线程内前面加过后面不用加）
     *      2、redisson 会自动解锁吗？
     *         会：看门狗30s。代码闪断，定时线程没有，就不自动续期了，等30s，redis自己删除
     *         默认加锁就是30s
     *      3、推荐不要使用 lock(30,TimeUnit.SECONDS)；失去了自动续期功能、
     *        scheduleExpirationRenewal(threadId); 的调用前提 是 if (leaseTime == -1)
     *        没有续期功能;
     *
     * 看门狗为什么？只是为了业务超时，锁时间不够来不停续期的
     *  看门狗还有默认的30秒（lock.lock()）自动过期（redis删），就是防止特殊情况
     *  类似断电 可以正常 释放锁 避免造成死锁吗
     *
     * 看门狗是续期的。每隔10s续满期。
     *
     *
     *
     * 2、redisson分布式业务超时怎么办？
     *      1、看门狗自动续期，每隔1/3看门狗时间就续满看门狗时间
     *
     * @param skuId
     * @return
     */
    @SneakyThrows  //使用redisson分布式锁做的方法
    public Map<String, Object> getSkuInfoWithRedissonDistributeLock(Long skuId) {

        //查50 100w  sku:info:50    lock:50
        //查70 100w  sku:info:70    lock:70
        //RLock lock = redissonClient.getLock("lock"),这样锁了200w
        //锁的粒度应该精细到一条记录。类似mysql的行锁

        log.info("准备查询" + skuId + " 号数据");
//        Lock lock1 = new ReentrantLock();
//        lock1.lock();
        //这段代码是加锁的吗？
        RLock lock = redissonClient.getLock("lock:" + skuId); //把锁对象拿到
        /**
         * //1、原生方式
         * locked: 默认传false
         * beforeToken默认传null
         */
//        return getSkuInfoWithRedisLock01(skuId,false,null);
//        RFuture<Void> voidRFuture = lock.lockAsync(); //异步加锁
//        System.out.println("aaa");
//
//        //监听，异步回调
//        voidRFuture.onComplete((v,e)->{
//            //业务代码
//            System.out.println("bbb");
//        });

        //2、redisson版
        Map<String, Object> cache = queryFromCache(skuId);
        if(cache == null){
            log.info("redis缓存没命中...准备查询数据库...尝试加锁");
            //3、加锁
            try{
                //看门狗，用来自动续期,redisson的所有功能都是原子性的
                lock.lock();  //这是一个自旋,可以续期，锁的默认时间是30 * 1000; 30s
                //每隔10s定时任务会自动续满期 internalLockLeaseTime / 3，
//                lock.lock(30,TimeUnit.SECONDS);  //30秒以后自动解锁，不会自动续期
                log.info("加锁成功....尝试继续命中缓存....");
                Map<String, Object> cacheAgain = queryFromCache(skuId);
                if(cacheAgain == null){
                    log.info("开始查询数据库....");
                    Thread.sleep(300);
                    Map<String, Object> data = getFromServiceItemFeign(skuId);
                    saveToCache(data,skuId);
                    return data;
                }

                log.info("命中缓存，直接返回....");
                return cacheAgain;
            }finally {
                //redisson 感知到解的不是自己的锁，然后就会抛出异常
                lock.unlock(); //解锁代码由于前面断电？
            }
        }
        log.info("缓存命中，直接返回.....");
        return cache;
    }

    /**
     * 如果某个订单模块的集群超过数据库并发数，是不是需要分布式锁？那前面需要再加上【布隆过滤器】吗？
     * 1、有缓存就加 【布隆过滤器】维护布隆过滤器也不容易，【每天晚上定时任务重建布隆过滤器】
     * 2、数据库并发数（....），
     *    各种业务应该有自己的布隆过滤器，每个布隆都很小，效率很高
     * AOP+注解完成的
     *    分布式布隆过滤防击穿高性能本地锁数据缓存切面
     *
     * 缓存空值
     * @param skuId
     * @return
     */
    @GmallCache(
            cacheKeyExpr = RedisConst.SKUKEY_PREFIX + "#{#args[0]}" + RedisConst.SKUKEY_SUFFIX,
            ttl = 1000 * 60 * 30,
            bloomName = BloomName.SKU, //用哪个布隆过滤器可以自己指定
            missDataTtl = 1000 * 60 * 10
    )  //需要一个切面类，动态切入所有标了这个注解的方法
    @Override
    //    @Cacheable(cacheNames = "sku",key = "#{#root.args[0]+':info'}") //类似缓存的分区，  sku:xxxxx
    public Map<String, Object> getSkuInfo(Long skuId) {
        //查数据
        log.info("要查数据库了.......");
        Map<String, Object> map = getFromServiceItemFeign(skuId);
        return map;
    }

    /**
     * 延迟更新，
     * @param skuId
     */
    @Override
    public void updateSkuHotScore(Long skuId) {
        //每次点击商品都更新热度
        // 100w    100  当redis  skuId :  200
        CompletableFuture.runAsync(()->{
            ZSetOperations<String, String> zset = stringRedisTemplate.opsForZSet();
            //只是给redis里面加值
            Double hotScore = zset.incrementScore("hotScore", skuId.toString(), 1);
            if(hotScore % 10 == 0){
                skuEsFeignClient.updateSkuHotScore(skuId,hotScore.longValue());
            }
        },executor);
    }

    //    @GmallCacheEvict()   //把指定的数据清除  prefix + skuInfo.skuId + suffix
//    public void uodate(SkuInfo skuInfo){
//        //灵活，方法修改了数据，手动清除缓存
//    }


    /**
     * 远程调用 service-product 查询出和当前 skuId 对应的所有信息
     * 缓存的使用逻辑是固定，我们完全可以抽取
     * <p>
     * 就可以使用AOP直接抽取出来,数据改了，缓存也要改？  如何使用SpringCache简化缓存开发
     *
     * @param skuId
     * @return 1、百万并发进来，正好skuId=50号数据过期
     * <p>
     * synchronized: 什么锁？ 对象锁，这个锁用的是 ItemServiceImpl 对象，this
     * 只要所有人啥都共用一把锁就能锁住
     * 能：因为 ItemServiceImpl 单实例的。大家进来都是同一个，会大，资源也有消耗
     * service有一个峰值： 4000/s
     * <p>
     * juc的所有锁都是本地锁，只针对当前应用有效，分布式情况锁不住
     */

//    @Override
//    public Map<String, Object> getSkuInfo(Long skuId) {
//        //        new xxx()
////        new Thread(xxxx)
////        new Thread(xxxx)
////        new Thread(xxxx)
////        new Thread(xxxx)
//        //只需要让所有人用同一把锁就能锁住
//        //场景： 数据库MySQL：   修改一条数据   1000   update xxx where id=5
//        //锁可以是数据库。
//        //大家都去一个地方占位，能占到说明拿到锁。这个位有东西，那就说明别人占用了锁，我们就稍等
//        //最快的就是redis
//        //1、test_lock    id   value
//        //                1    xxxxx
//        //2、minio。存一个同名文件。只要我存进来，别人一看有就不存了。
//        //3、分布式中间件，谁都能占坑。mq。create queue。
////        synchronized (ItemServiceImpl.class){
////            //效果一样,Spring原因，this已经是单例锁了。
////            //同一个jvm上部署
////            //分布式下。应用在多个机器
////        }
//
//        ObjectMapper mapper = new ObjectMapper();
//        /**
//         * 1、所有查询之前，先看缓存
//         *   1.1）、如果缓存中有就使用缓存的
//         *   1.2）、如果缓存没有，就查数据库
//         */
//        ValueOperations<String, String> operations =
//                stringRedisTemplate.opsForValue();
//
//        //  50 51  52   sku:info:51    sku:info:52   sku:info:53
//        //1、先去缓存确定是否存在
//        String redisContent = operations.get("sku:info:" + skuId);
//
//
//
//
//        if(StringUtils.isEmpty(redisContent)){
//            //分布式锁
//             //String   ok =   set a b NX
////            if(ok){
////                //缓存中没有，远程查询
////                Map<String, Object> fromServiceItemFeign = getFromServiceItemFeign(skuId);
////            }else{
////              //自旋锁
////             while((set a b NX).equals("ok")){
////                  getSkuInfo(50);
////             }
////            }
//
//            //2、缓存中没有，远程查询
//            Map<String, Object> fromServiceItemFeign = getFromServiceItemFeign(skuId);
//            //请存到磁盘，存到远程？？？？
////            ObjectOutputStream objectOutputStream = new ObjectOutputStream();
////            //java的序列化
////            objectOutputStream.writeObject(fromServiceItemFeign);
//
//            String jsonStr = null;
//            try {
//                //序列化......
//                jsonStr = mapper.writeValueAsString(fromServiceItemFeign);
//            } catch (JsonProcessingException e) {
//                e.printStackTrace();
//            }
//            //2.1）、给redis中存一份
//            operations.set("sku:info:" + skuId,jsonStr);
//
//            //2.2）、返回数据
//            return fromServiceItemFeign;
//        }else{
//            Map<String, Object> stringObjectMap = null;
//            try {
//                //redis拿到的是str还要转对象   反序列化
//                stringObjectMap = mapper.readValue(redisContent, new TypeReference<Map<String, Object>>() {
//                });
//            } catch (JsonProcessingException e) {
//                e.printStackTrace();
//            }
//            return stringObjectMap;
//        }
//        //把redis锁释放
//    }


    private Map<String, Object> getFromServiceItemFeign(Long skuId){
        Map<String, Object> result = new HashMap<>();
        //我以下的写法，
        // 对比  new Thread(()->{}).start(); 优点：
        //1、线程重复使用。 16~32直接重复使用，少了开线程时间  new Thread  0.01
        //2、线程吞吐量的限制。
        // 以前 1个请求 5个线程，   100请求，new 500个线程   1w个请求  5w线程等待CPU切换（内存占用更大）
        // 现在 1个请求 5个线程，交给线程池，线程池只有32个一直执行。控制资源
        //     100 请求 500个线程，交给线程池， 之前32个线程等待CPU切换，  468 就在队列等待执行
        //     1w 请求  1w个线程，交给线程池，之前32个线程等待CPU切换，  9968 个在队列（占内存）

        //1、查询sku详情
        CompletableFuture<SkuInfo> future = CompletableFuture.supplyAsync(() -> {
            SkuInfo skuInfo = skuInfoFeignClient.getSkuInfo(skuId);
            result.put("skuInfo", skuInfo);
            return skuInfo;
        }, executor);

        //1~4:可交给线程池并行执行
        //1、查询分类
        CompletableFuture<Void> future1 = future.thenAcceptAsync((res) -> {
            if (res != null) {
                BaseCategoryView skuCategorys = skuInfoFeignClient.getSkuCategorys(res.getCategory3Id());
                result.put("categoryView", skuCategorys);
            }
        }, executor);


        //2、查询销售属性
        CompletableFuture<Void> future2 = future.thenAcceptAsync((res) -> {
            if (res != null) {
                List<SpuSaleAttr> spuSaleAttrListCheckBySku = skuInfoFeignClient.getSpuSaleAttrListCheckBySku(skuId, res.getSpuId());
                result.put("spuSaleAttrList", spuSaleAttrListCheckBySku);
            }
        }, executor);

        //3、查询价格
        CompletableFuture<Void> future3 = future.thenAcceptAsync((res) -> {
            if (res != null) {
                BigDecimal skuPrice = skuInfoFeignClient.getSkuPrice(skuId);
                result.put("price", skuPrice);
            }
        }, executor);

        //4、查询组合
        CompletableFuture<Void> future4 = future.thenAcceptAsync((res) -> {
            if (res != null) {
                Map map = skuInfoFeignClient.getSkuValueIdsMap(res.getSpuId());
                ObjectMapper mapper = new ObjectMapper();
                try {
                    String jsonStr = mapper.writeValueAsString(map);
                    log.info("valuesSkuJson 内容：{}", jsonStr);
                    result.put("valuesSkuJson", jsonStr);
                } catch (JsonProcessingException e) {
                    log.error("商品sku组合数据转换异常：{}", e);
                }
            }
        }, executor);

        CompletableFuture<Void> allOf = CompletableFuture.allOf(future, future1, future2, future3, future4);

        try {
            allOf.get();
        } catch (Exception e) {
            log.error("线程池异常：{}",e);
        }

        return result;
    }

    /**
     * 远程查询sku详细信息
     * @param skuId
     * @return
     */
    private Map<String, Object> getFromServiceItemFeign01(Long skuId) {
        //        Thread thread = new Thread();
//        Thread thread1 = new Thread();
//        Thread thread2 = new Thread();
//
//        //线程没有顺序
//        thread.start();
//        thread1.start();
//        thread2.start();
        //异步任务既能异步，还有按序编排
        //1、查Sku图片，基本信息    0.2s  //五个人都异步来做
        //2、查Sku的目录           0.2s
        //3、查Sku的销售属性       0.2s
        //4、查Sku的价格          0.2s
        //5、查Sku属性组合信息     0.2s

        //串行执行。执行完需要  1s； 影响吞吐量
        //   1s 能接  共计 500 请求
        //tomcat线程池最大500；
        //同步：  1个请求阻塞1s， 500请求同时进来。 500把tomcat塞满。 并发就是 500
        //异步：  1个请求阻塞0.2s， 500请求同时进来， 500只能把tomcat占用0.2s， 并发就是2500  吞吐量
        // 异步==快？   异步提升吞吐量的  总时间是一样的。

        //1、查出订单详情。订单详情中有物流单号
        //2、根据物流单号，查询物流信息
        //3、根据物流单号，查询仓库信息
        //4、根据物流单号，查询xxx信息



        log.info("开始远程查询，远程会操作数据库.....");
        Map<String, Object> result = new HashMap<>();
        //skuInfo信息

        //RPC 查询  skuDetail
        //1、Sku基本信息（名字，id，xxx，价格，sku_描述） sku_info
        //2，Sku图片信息（sku的默认图片[sku_info]，sku_image[一组图片]）
        SkuInfo skuInfo = skuInfoFeignClient.getSkuInfo(skuId);
        if (skuInfo != null) {
            result.put("skuInfo", skuInfo);
            //3，Sku分类信息（sku_info[只有三级分类]，根据这个三级分类查出所在的一级，二级分类内容，连上三张分类表继续查）
            BaseCategoryView skuCategorys = skuInfoFeignClient.getSkuCategorys(skuInfo.getCategory3Id());
            result.put("categoryView", skuCategorys);
            //4，Sku销售属性相关信息（查出自己的sku组合，还要查出这个sku所在的spu定义了的所有销售属性和属性值）
            List<SpuSaleAttr> spuSaleAttrListCheckBySku = skuInfoFeignClient.getSpuSaleAttrListCheckBySku(skuId, skuInfo.getSpuId());
            result.put("spuSaleAttrList", spuSaleAttrListCheckBySku);

            //5，Sku价格信息（平台可以单独修改价格，sku后续会放入缓存，为了回显最新价格，所以单独获取）
            BigDecimal skuPrice = skuInfoFeignClient.getSkuPrice(skuId);
            result.put("price", skuPrice);
            //6、SPU下面的所有存在的sku组合信息 {"121|123|156":65,"122|123|111":67}
            //前端这里还需要把map转成json字符串， JSON.parse(.valuesSkuJson)
            Map map = skuInfoFeignClient.getSkuValueIdsMap(skuInfo.getSpuId());
            ObjectMapper mapper = new ObjectMapper();
            try {
                String jsonStr = mapper.writeValueAsString(map);
                log.info("valuesSkuJson 内容：{}", jsonStr);
                result.put("valuesSkuJson", jsonStr);
            } catch (JsonProcessingException e) {
                log.error("商品sku组合数据转换异常：{}", e);
            }

        }
        return result;
    }
}
